package com.shujia.sql

import org.apache.spark.sql.{Column, DataFrame, SaveMode, SparkSession}

/**
 * 1、行列转换
 *
 * 表1
 * 姓名,科目,分数
 * name,item,score
 * 张三,数学,33
 * 张三,英语,77
 * 李四,数学,66
 * 李四,英语,78
 *
 *
 * 表2
 * 姓名,数学,英语
 * name,math,english
 * 张三,33,77
 * 李四,66,78
 *
 * 1、将表1转化成表2
 * 2、将表2转化成表1
 */
object Demo10Student {
  def main(args: Array[String]): Unit = {
    //创建SparkSession对象
    /**
     * 在新版本的spark中，如果想要编写spark sql的话，需要使用新的spark入口类：SparkSession
     */
    val sparkSession: SparkSession = SparkSession.builder()
      .master("local")
      .appName("行列互相转换演示")
      .config("spark.sql.shuffle.partitions", 1) //默认分区的数量是200个
      .getOrCreate()

    //导入Spark sql中所有的sql隐式转换函数
    import org.apache.spark.sql.functions._
    //导入另一个隐式转换，后面可以直接使用$函数引用字段进行处理,如果需要做RDD和DF之间的转换
    import sparkSession.implicits._

    /**
     * 列转行
     */
//    val tb1DF: DataFrame = sparkSession.read
//      .format("csv")
//      .option("sep", ",")
//      .schema("name STRING,item STRING,score INT")
//      .load("/bigdata29/tb1.txt")

    /*
    张三,数学,33
    张三,英语,77
    李四,数学,66
    李四,英语,78

    张三,33,77
    李四,66,78
     */
//    val res1DF: DataFrame = tb1DF.groupBy($"name")
//      .agg(
//        sum(when($"item" === "数学", $"score").otherwise(0)) as "math",
//        sum(when($"item" === "英语", $"score").otherwise(0)) as "english")
//
//    res1DF.write
//      .format("csv")
//      .option("sep",",")
//      .mode(SaveMode.Overwrite)
//      .save("/bigdata29/out7")

    /**
     * 行转列
     */
    val tb2DF: DataFrame = sparkSession.read
      .format("csv")
      .option("sep", ",")
      .schema("name STRING,math STRING,english INT")
      .load("/bigdata29/out7/*")

    val m: Column = map(
      expr("'数学'"), $"math",
      expr("'语文'"), $"english"
    )
    tb2DF.select($"name",explode(m) as Array("item","score")).show()
  }
}
